package com.example.rabbitmq.starter.service;

import com.example.rabbitmq.starter.dao.FailedMessageRepository;
import com.example.rabbitmq.starter.entity.FailedMessageInfo;
import com.example.rabbitmq.starter.entity.MQConfigInfo;
import com.example.rabbitmq.starter.handler.FailureHandlerFactory;
import com.example.rabbitmq.starter.model.MessagePayload;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import lombok.RequiredArgsConstructor;

import javax.annotation.Resource;
import javax.transaction.Transactional;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.UUID;

/**
 * 失败消息服务
 * 负责管理失败消息的查询、重试、忽略、删除等操作
 *
 * @author example
 * @since 1.0.0
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class FailedMessageService {

    private final FailedMessageRepository failedMessageRepository;

    private final RabbitTemplate rabbitTemplate;

    private final RabbitAdmin rabbitAdmin;

    private final ObjectMapper objectMapper;

    private final MQConfigService mqConfigService;

    private final FailureHandlerFactory failureHandlerFactory;

    /**
     * 查询所有失败消息
     *
     * @return 失败消息列表
     */
    public List<FailedMessageInfo> getAllFailedMessages() {
        return failedMessageRepository.findAll();
    }

    /**
     * 分页查询所有失败消息
     *
     * @param pageable 分页参数
     * @return 分页结果
     */
    public Page<FailedMessageInfo> findAll(Pageable pageable) {
        return failedMessageRepository.findAll(pageable);
    }


    /**
     * 根据消息类型查询失败消息（分页）
     *
     * @param messageType 消息类型
     * @param pageable    分页参数
     * @return 分页结果
     */
    public Page<FailedMessageInfo> findByMessageType(String messageType, Pageable pageable) {
        return failedMessageRepository.findByMessageType(messageType, pageable);
    }

    /**
     * 根据状态查询失败消息（分页）
     *
     * @param status   状态
     * @param pageable 分页参数
     * @return 分页结果
     */
    public Page<FailedMessageInfo> findByStatus(String status, Pageable pageable) {
        return failedMessageRepository.findByStatus(status, pageable);
    }

    /**
     * 根据消息类型和状态查询失败消息（分页）
     *
     * @param messageType 消息类型
     * @param status      状态
     * @param pageable    分页参数
     * @return 分页结果
     */
    public Page<FailedMessageInfo> findByMessageTypeAndStatus(String messageType, String status, Pageable pageable) {
        return failedMessageRepository.findByMessageTypeAndStatus(messageType, status, pageable);
    }

    /**
     * 根据消息类型查询失败消息（非分页）
     *
     * @param messageType 消息类型
     * @return 失败消息列表
     */
    public List<FailedMessageInfo> findByMessageType(String messageType) {
        return failedMessageRepository.findByMessageType(messageType);
    }

    /**
     * 根据状态查询失败消息（非分页）
     *
     * @param status 状态
     * @return 失败消息列表
     */
    public List<FailedMessageInfo> findByStatus(String status) {
        return failedMessageRepository.findByStatus(status);
    }

    /**
     * 根据消息类型和状态查询失败消息（非分页）
     *
     * @param messageType 消息类型
     * @param status      状态
     * @return 失败消息列表
     */
    public List<FailedMessageInfo> findByMessageTypeAndStatus(String messageType, String status) {
        return failedMessageRepository.findByMessageTypeAndStatus(messageType, status);
    }

    /**
     * 查询所有失败消息（非分页）
     *
     * @return 失败消息列表
     */
    public List<FailedMessageInfo> findAll() {
        return failedMessageRepository.findAll();
    }

    /**
     * 根据ID查询失败消息
     *
     * @param id 失败消息ID
     * @return 失败消息信息
     */
    public FailedMessageInfo getFailedMessageById(String id) {
        Optional<FailedMessageInfo> optional = failedMessageRepository.findById(id);
        return optional.orElse(null);
    }

    /**
     * 分页查询失败消息
     *
     * @param page        页码（从0开始）
     * @param size        每页大小
     * @param messageType 消息类型（可选）
     * @param status      状态（可选）
     * @param businessId  业务ID（可选）
     * @return 分页结果
     */
    public Page<FailedMessageInfo> getFailedMessagesPage(int page, int size, String messageType, String status, String businessId) {
        Pageable pageable = PageRequest.of(page, size, Sort.by(Sort.Direction.DESC, "failTime"));

        // 根据条件查询
        if (StringUtils.hasText(messageType) && StringUtils.hasText(status)) {
            return failedMessageRepository.findByMessageTypeAndStatus(messageType, status, pageable);
        }
        else if (StringUtils.hasText(messageType)) {
            return failedMessageRepository.findByMessageType(messageType, pageable);
        }
        else if (StringUtils.hasText(status)) {
            return failedMessageRepository.findByStatus(status, pageable);
        }
        else if (StringUtils.hasText(businessId)) {
            return failedMessageRepository.findByBusinessId(businessId, pageable);
        }
        else {
            return failedMessageRepository.findAll(pageable);
        }
    }

    /**
     * 重试失败消息
     *
     * @param id 失败消息ID
     * @return 重试结果
     */
    @Transactional
    public boolean retryMessage(String id) {
        Optional<FailedMessageInfo> optionalMessage = failedMessageRepository.findById(id);
        if (!optionalMessage.isPresent()) {
            log.warn("重试失败消息不存在：id={}", id);
            return false;
        }

        FailedMessageInfo failedMessage = optionalMessage.get();

        try {
            // 重新构建消息
            Message message = rebuildMessage(failedMessage);

            // 发送消息
            rabbitTemplate.send(failedMessage.getExchangeName(), failedMessage.getRoutingKey(), message);

            // 重试成功后直接删除消息记录
            failedMessageRepository.deleteById(id);
            log.info("重试失败消息成功：id={}, messageId={}", id, failedMessage.getMessageId());
            return true;
        }
        catch (Exception e) {
            log.error("重试失败消息异常：id={}, error={}", id, e.getMessage(), e);
            // 更新状态
            Date now = new Date();
            failedMessageRepository.updateStatus(id, "RETRY_FAILED", "重试失败：" + e.getMessage(), now);
            return false;
        }
    }

    /**
     * 批量重试失败消息
     *
     * @param ids 失败消息ID列表
     * @return 重试成功的消息数量
     */
    public int retryFailedMessages(List<String> ids) {
        int successCount = 0;

        for (String id : ids) {
            if (retryMessage(id)) {
                successCount++;
            }
        }

        return successCount;
    }

    /**
     * 重试所有失败消息
     *
     * @return 重试成功的消息数量
     */
    public int retryAllFailedMessages() {
        // 获取所有待处理的失败消息
        List<FailedMessageInfo> pendingMessages = failedMessageRepository.findByStatus("PENDING");
        int successCount = 0;

        if (pendingMessages.isEmpty()) {
            log.info("没有需要重试的失败消息");
            return 0;
        }

        log.info("开始重试所有失败消息，共 {} 条", pendingMessages.size());

        for (FailedMessageInfo message : pendingMessages) {
            if (retryMessage(message.getId())) {
                successCount++;
            }
        }

        log.info("重试失败消息完成，成功: {}, 总数: {}", successCount, pendingMessages.size());
        return successCount;
    }

    /**
     * 忽略失败消息
     *
     * @param id     失败消息ID
     * @param reason 忽略原因
     * @return 操作结果
     */
    @Transactional
    public boolean ignoreMessage(String id, String reason) {
        Optional<FailedMessageInfo> optionalMessage = failedMessageRepository.findById(id);
        if (!optionalMessage.isPresent()) {
            log.warn("忽略失败消息不存在：id={}", id);
            return false;
        }

        // 更新状态
        Date now = new Date();
        int updated = failedMessageRepository.updateStatus(id, "IGNORED", reason, now);

        log.info("忽略失败消息：id={}, reason={}", id, reason);
        return updated > 0;
    }

    /**
     * 批量忽略失败消息
     *
     * @param ids    失败消息ID列表
     * @param reason 忽略原因
     * @return 操作成功的消息数量
     */
    public int ignoreFailedMessages(List<String> ids, String reason) {
        int successCount = 0;

        for (String id : ids) {
            if (ignoreMessage(id, reason)) {
                successCount++;
            }
        }

        return successCount;
    }

    /**
     * 删除失败消息
     *
     * @param id 失败消息ID
     * @return 操作结果
     */
    @Transactional
    public boolean deleteMessage(String id) {
        Optional<FailedMessageInfo> optionalMessage = failedMessageRepository.findById(id);
        if (!optionalMessage.isPresent()) {
            log.warn("删除失败消息不存在：id={}", id);
            return false;
        }

        // 删除数据库记录
        failedMessageRepository.deleteById(id);

        log.info("删除失败消息成功：id={}", id);
        return true;
    }

    /**
     * 批量删除失败消息
     *
     * @param ids 失败消息ID列表
     * @return 操作成功的消息数量
     */
    public int deleteFailedMessages(List<String> ids) {
        int successCount = 0;

        for (String id : ids) {
            if (deleteMessage(id)) {
                successCount++;
            }
        }

        return successCount;
    }


    /**
     * 保存失败消息
     *
     * @param messageInfo 失败消息信息
     * @return 保存后的失败消息信息
     */
    public FailedMessageInfo saveFailedMessage(FailedMessageInfo messageInfo) {
        if (messageInfo.getId() == null || messageInfo.getId().isEmpty()) {
            messageInfo.setId(UUID.randomUUID().toString());
        }

        if (messageInfo.getFailTime() == null) {
            messageInfo.setFailTime(new Date());
        }

        return failedMessageRepository.save(messageInfo);
    }

    /**
     * 从失败消息信息重新构建消息对象
     *
     * @param failedMessage 失败消息信息
     * @return 消息对象
     */
    private Message rebuildMessage(FailedMessageInfo failedMessage) throws JsonProcessingException {
        // 解析原始消息内容
        MessagePayload messagePayload = objectMapper.readValue(failedMessage.getMessageContent(), MessagePayload.class);

        // 重置重试次数为0
        messagePayload.setRetryCount(0);

        // 构建消息属性
        MessageProperties props = new MessageProperties();
        props.setMessageId(failedMessage.getMessageId());
        props.setType(failedMessage.getMessageType());
        props.setContentType(MessageProperties.CONTENT_TYPE_JSON);
        props.setTimestamp(new Date());

        // 设置消息体（使用重置后的MessagePayload）
        byte[] body = objectMapper.writeValueAsString(messagePayload).getBytes();

        // 构建消息对象
        return new Message(body, props);
    }

    /**
     * 保存失败消息（从MessagePayload）
     *
     * @param messagePayload 消息载体
     * @param errorMessage   错误信息
     * @return 保存的失败消息信息
     */
    public FailedMessageInfo persistFailedMessage(MessagePayload messagePayload, String errorMessage) {
        try {
            // 检查消息是否已存在
            FailedMessageInfo existingMessage = failedMessageRepository.findByMessageId(messagePayload.getId());
            if (existingMessage != null) {
                log.info("失败消息已存在，更新重试次数：messageId={}", messagePayload.getId());
                // 增加重试次数
                failedMessageRepository.incrementRetryCount(existingMessage.getId(), new Date());
                return existingMessage;
            }

            // 创建失败消息记录
            FailedMessageInfo failedMessage = new FailedMessageInfo();
            failedMessage.setId(UUID.randomUUID().toString());
            failedMessage.setMessageId(messagePayload.getId());
            failedMessage.setMessageType(messagePayload.getType());
            failedMessage.setBusinessId(messagePayload.getBusinessId());
            failedMessage.setMessageContent(objectMapper.writeValueAsString(messagePayload));
            failedMessage.setFailReason(errorMessage);
            failedMessage.setFailTime(new Date());
            failedMessage.setRetryCount(messagePayload.getRetryCount());
            failedMessage.setStatus("PENDING");

            // 获取MQ配置
            MQConfigInfo config = mqConfigService.getConfigByMessageType(messagePayload.getType());
            if (config != null) {
                failedMessage.setExchangeName(config.getExchangeName());
                failedMessage.setRoutingKey(config.getRoutingKey());
                failedMessage.setQueueName(config.getQueueName());
                failedMessage.setMaxRetryCount(config.getMaxRetryCount());
            }
            else {
                failedMessage.setMaxRetryCount(messagePayload.getMaxRetryCount());
            }

            // 构建消息对象用于持久化到文件
            MessageProperties properties = MessagePropertiesBuilder.newInstance()
                    .setContentType(MessageProperties.CONTENT_TYPE_JSON)
                    .setMessageId(messagePayload.getId())
                    .setType(messagePayload.getType())
                    .setTimestamp(new Date())
                    .build();

            // 保存到数据库
            FailedMessageInfo savedMessage = failedMessageRepository.save(failedMessage);

            // 检查是否达到最大重试次数，如果是则调用失败处理器
            checkAndExecuteFailureHandler(savedMessage, messagePayload, new RuntimeException(errorMessage));

            return savedMessage;
        }
        catch (Exception e) {
            log.error("保存失败消息异常：error={}", e.getMessage(), e);
            return null;
        }
    }

    /**
     * 检查并执行失败处理器
     *
     * @param failedMessage  失败消息信息
     * @param messagePayload 消息载体
     * @param lastException  最后一次异常
     */
    private void checkAndExecuteFailureHandler(FailedMessageInfo failedMessage, MessagePayload messagePayload, Exception lastException) {
        try {
            // 检查是否达到最大重试次数
            int currentRetryCount = messagePayload.getRetryCount();
            int maxRetryCount = failedMessage.getMaxRetryCount();

            if (currentRetryCount >= maxRetryCount) {
                log.info("消息已达到最大重试次数，准备执行失败处理器: messageId={}, retryCount={}, maxRetryCount={}",
                        messagePayload.getId(), currentRetryCount, maxRetryCount);

                // 获取MQ配置中的失败处理器类路径
                MQConfigInfo config = mqConfigService.getConfigByMessageType(messagePayload.getType());
                if (config != null && StringUtils.hasText(config.getFailureHandlerClass())) {
                    // 异步执行失败处理器，避免阻塞主流程
                    executeFailureHandlerAsync(config.getFailureHandlerClass(), failedMessage, messagePayload, lastException);
                }
                else {
                    log.warn("未配置失败处理器或配置为空: messageType={}, messageId={}",
                            messagePayload.getType(), messagePayload.getId());
                }
            }
        }
        catch (Exception e) {
            log.error("检查和执行失败处理器时发生异常: messageId={}, error={}",
                    messagePayload.getId(), e.getMessage(), e);
        }
    }

    /**
     * 异步执行失败处理器
     *
     * @param handlerClassName 处理器类名
     * @param failedMessage    失败消息信息
     * @param messagePayload   消息载体
     * @param lastException    最后一次异常
     */
    private void executeFailureHandlerAsync(String handlerClassName, FailedMessageInfo failedMessage,
                                            MessagePayload messagePayload, Exception lastException) {
        // 使用新线程异步执行，避免阻塞消息处理流程
        new Thread(() -> {
            try {
                failureHandlerFactory.executeFailureHandler(
                        handlerClassName,
                        messagePayload.getId(),
                        failedMessage.getMessageContent(),
                        messagePayload.getType(),
                        messagePayload.getRetryCount(),
                        lastException
                );
            }
            catch (Exception e) {
                log.error("异步执行失败处理器时发生异常: messageId={}, handlerClass={}, error={}",
                        messagePayload.getId(), handlerClassName, e.getMessage(), e);
            }
        }, "FailureHandler-" + messagePayload.getId()).start();
    }

    /**
     * 根据消息类型获取MQ配置
     *
     * @param messageType 消息类型
     * @return MQ配置信息
     */
    public MQConfigInfo getMQConfigByMessageType(String messageType) {
        return mqConfigService.getConfigByMessageType(messageType);
    }


}